package com.hliushi.process

import com.alibaba.fastjson.JSON
import com.hliushi.bean.{CovidBean, StatisticsDataBean}
import com.hliushi.utils.{BasicJdbcSink, GlobalConfigUtil}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.{Row, SparkSession}

/**
 * descriptions: 全国各省市疫情数据实时处理与分析
 *
 * author: Hliushi
 * date: 2021/7/1 22:46
 */
object Covid19_Data_Process {

  /**
   * Structured Streaming 采用了无界表的概念, 流处理相当于往一个表上不断追加行
   * 基于Spark SQL引擎实现, 可以使用大多数Spark SQL的function
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {
    // 1.创建StructuredStreaming执行环境
    // StructuredStreaming支持使用SQL来处理实时流数据, 数据抽象和SparkSQL一样, 也是DataFrame和DataSet
    // 所以这里创建StructuredStreaming执行环境就直接创建SparkSession即可
    val spark = SparkSession.builder().master("local[6]")
      .appName("covid19_data_process")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.setLogLevel("WARN")
    // 导入隐式转换方便后续使用
    import org.apache.spark.sql.functions._
    import spark.implicits._

    import scala.collection.JavaConverters._

    // 2.连接Kafka
    // 从kafka接收信息
    val kafkaDF = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", GlobalConfigUtil.bootstrapServers)
      .option("subscribe", "covid19")
      .load()

    // 取出消息中的Value
    val jsonStrDS = kafkaDF.selectExpr("CAST(value AS STRING)").as[String]
    //jsonStrDS.writeStream
    //  .format("console") // 输出目的地
    //  .outputMode("append") // 输出模式, 默认就是append, 表示显示新增行
    //  .trigger(Trigger.ProcessingTime(0)) // 触发间隔, 0表示尽可能快的执行
    //  .option("truncate", "false") // 表示如果列名过长, 不进行截断
    //  .start()
    //  .awaitTermination()

    //{"confirmedCount":6,"curedCount":6,"currentConfirmedCount":0,"datetime":"2021-07-02","deadCount":0,"highDangerCount":0,"locationId":441500,"midDangerCount":0,"pid":440000,"provinceShortName":"广东","suspectedCount":0}
    //|{"confirmedCount":5,"curedCount":5,"currentConfirmedCount":0,"datetime":"2021-07-02","deadCount":0,"highDangerCount":0,"locationId":441600,"midDangerCount":0,"pid":440000,"provinceShortName":"广东","suspectedCount":0}

    // 3.处理数据
    // 将jsonStr转为样例类
    val covidBeanDS = jsonStrDS.map((jsonStr: String) => {
      // 注意:Scala中获取class对象使用 classOf[类名]
      // Java中使用类名.class / Class.forName(全类路径) / 对象.getClass
      JSON.parseObject(jsonStr, classOf[CovidBean])
    })
    // 分离出省份数据
    val provinceDS = covidBeanDS.filter(x => StringUtils.isNotBlank(x.statisticsData))

    // 分离出城市数据
    val cityDS = covidBeanDS.filter(x => StringUtils.isBlank(x.statisticsData))

    // 分离出各省份每一天的统计数据
    val statisticsDataDS = provinceDS.flatMap((p: CovidBean) => {
      // 获取到的是该省份每一天的统计数据组成的jsonStr数组
      val jsonStr = p.statisticsData
      val list = JSON.parseArray(jsonStr, classOf[StatisticsDataBean]).asScala
      val newList = list.map((s: StatisticsDataBean) => {
        s.provinceShortName = p.provinceShortName
        s.locationId = p.locationId
        s
      })
      newList
    })

    //statisticsDataDS.writeStream
    //  .format("console") // 输出目的地
    //  .outputMode("append") // 输出模式, 默认就是append, 表示显示新增行
    //  .trigger(Trigger.ProcessingTime(0)) // 触发间隔, 0表示尽可能快的执行
    //  .option("truncate", "false") // 表示如果列名过长, 不进行截断
    //  .start()
    //  .awaitTermination()

    //+-----------------+----------+--------+--------------+-------------+----------+---------+---------------------+--------------------+---------+--------+---------------+--------------+--------------+------------------+
    //|provinceShortName|locationId|dateId  |confirmedCount|confirmedIncr|curedCount|curedIncr|currentConfirmedCount|currentConfirmedIncr|deadCount|deadIncr|highDangerCount|midDangerCount|suspectedCount|suspectedCountIncr|
    //+-----------------+----------+--------+--------------+-------------+----------+---------+---------------------+--------------------+---------+--------+---------------+--------------+--------------+------------------+
    //|浙江               |330000    |20200121|5             |5            |0         |0        |5                    |5                   |0        |0       |0              |0             |0             |0                 |
    //|浙江               |330000    |20200122|10            |5            |0         |0        |10                   |5                   |0        |0       |0              |0             |0             |0                 |
    //|浙江               |330000    |20200123|43            |33           |1         |1        |42                   |32                  |0        |0       |0              |0             |0             |0                 |
    //|浙江               |330000    |20200124|62            |19           |1         |0        |61                   |19                  |0        |0       |0              |0             |0             |0                 |

    // 4.统计分析
    // 4.1.全国疫情汇总信息: 现有确诊, 累计确诊, 现有疑似, 累计治愈, 累计死亡 -- 注意:按照日期分组统计
    val result1 = provinceDS.groupBy($"datetime")
      .agg(sum($"currentConfirmedCount") as "currentConfirmedCount", // 现有确诊
        sum($"confirmedCount") as "confirmedCount", // 累计确诊
        sum($"suspectedCount") as "suspectedCount", // 现有疑似
        sum($"curedCount") as "curedCount", // 累计治愈
        sum($"deadCount") as "deadCount" // 累计死亡
      )

    // 4.2.全国个省份累计确诊数地图 -- 注意:按照日期-省份分组
    //cityDS.groupBy($"datetime", $"provinceShortName")
    //  .agg(sum($"confirmedCount") as "confirmedCount")
    val result2 = provinceDS.select($"datetime", $"locationId", $"provinceShortName", $"currentConfirmedCount", $"confirmedCount",
      $"suspectedCount", $"curedCount", $"deadCount")

    // 4.3.全国疫情趋势 -- 注意:按照日期分组聚合
    val result3 = statisticsDataDS.groupBy($"dateId")
      .agg(sum($"confirmedIncr") as "confirmedIncr", // 新增确诊
        sum($"confirmedCount") as "confirmedCount", // 累计确诊
        sum($"suspectedCount") as "suspectedCount", // 累计疑似
        sum($"curedCount") as "curedCount", // 累计治愈
        sum($"deadCount") as "deadCount" // 累计死亡
      )

    // 4.4.境外输入排行 -- 注意:按照日期-城市分组聚合
    val result4 = cityDS.filter(_.cityName.contains("境外输入"))
      .groupBy($"datetime", $"provinceShortName", $"pid")
      .agg(sum($"confirmedCount") as "confirmedCount")
      .sort($"confirmedCount".desc)

    // 4.5.统计北京市的累计确诊地图
    val result5 = cityDS.filter(x => x.provinceShortName.equals("北京"))
      .select($"datetime", $"locationId", $"provinceShortName", $"cityName", $"currentConfirmedCount",
        $"confirmedCount", $"suspectedCount", $"curedCount", $"deadCount")



    // 5.结果输出 -- 先输出到控制台观察, 最终输出到MySQL

    /**
     * .outputMode("complete")
     * 输出模式
     * 1.append: 默认的, 表示只输出新增的数据, 只支持简单的查询, 不支持聚合
     * 2.complete: 表示完整模式, 所有数据都会输出, 必须包含聚合操作
     * 3.update: 表示更新模式, 只输出有变化的数据, 不支持排序
     */
    //result1.writeStream
    //  .format("console")
    //  .outputMode("complete")
    //  .trigger(Trigger.ProcessingTime(0))
    //  .option("truncate", value = false)
    //  .start()
    ////.awaitTermination()
    // +----------+---------------------+--------------+--------------+----------+---------+
    // |datetime  |currentConfirmedCount|confirmedCount|suspectedCount|curedCount|deadCount|
    // +----------+---------------------+--------------+--------------+----------+---------+
    // |2021-07-02|4429                 |118644        |1979          |108707    |5508     |
    // +----------+---------------------+--------------+--------------+----------+---------+

    val result1Sql =
    """
      |replace into covid19_1 (`datetime`, `currentConfirmedCount`, `confirmedCount`, `suspectedCount`, `curedCount`, `deadCount`)
      |values (?, ?, ?, ?, ?, ?)
      |""".stripMargin

    result1.writeStream
      .foreach(new BasicJdbcSink(result1Sql) {
        override def realProcess(sql: String, row: Row): Unit = {
          val datetime = row.getAs[String]("datetime")
          val currentConfirmedCount = row.getAs[Long]("currentConfirmedCount")
          val confirmedCount = row.getAs[Long]("confirmedCount")
          val suspectedCount = row.getAs[Long]("suspectedCount")
          val curedCount = row.getAs[Long]("curedCount")
          val deadCount = row.getAs[Long]("deadCount")

          // 获取预编译语句对象
          ps = conn.prepareStatement(sql)
          // 给SQL设置参数值
          ps.setString(1, datetime)
          ps.setLong(2, currentConfirmedCount)
          ps.setLong(3, confirmedCount)
          ps.setLong(4, suspectedCount)
          ps.setLong(5, curedCount)
          ps.setLong(6, deadCount)

          ps.executeUpdate()
        }
      })
      .outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))
      .option("truncate", value = false)
      .start()


    //result2.writeStream
    //  .format("console")
    //  .outputMode("append")
    //  .trigger(Trigger.ProcessingTime(0))
    //  .option("truncate", value = false)
    //  .start()
    ////.awaitTermination()
    //+----------+----------+-----------------+---------------------+--------------+--------------+----------+---------+
    //|datetime  |locationId|provinceShortName|currentConfirmedCount|confirmedCount|suspectedCount|curedCount|deadCount|
    //+----------+----------+-----------------+---------------------+--------------+--------------+----------+---------+
    //|2021-07-02|110000    |北京              |12                   |1078          |164           |1057      |9        |
    //|2021-07-02|320000    |江苏              |11                   |743           |3             |732       |0        |
    //|2021-07-02|120000    |天津              |8                    |401           |50            |390       |3        |
    //|2021-07-02|430000    |湖南              |6                    |1057          |2             |1047      |4        |

    val result2Sql =
      """
        |replace into covid19_2 (`datetime`, `locationId`, `provinceShortName`, `currentConfirmedCount`,
        |`confirmedCount`, `suspectedCount`, `curedCount`, `deadCount`)
        |values (?, ?, ?, ?, ?, ?, ?, ?)
        |""".stripMargin

    result2.writeStream
      .foreach(new BasicJdbcSink(result2Sql) {
        override def realProcess(sql: String, row: Row): Unit = {
          val datetime = row.getAs[String]("datetime")
          val locationId = row.getAs[Int]("locationId")
          val provinceShortName = row.getAs[String]("provinceShortName")
          val currentConfirmedCount = row.getAs[Int]("currentConfirmedCount")
          val confirmedCount = row.getAs[Int]("confirmedCount")
          val suspectedCount = row.getAs[Int]("suspectedCount")
          val curedCount = row.getAs[Int]("curedCount")
          val deadCount = row.getAs[Int]("deadCount")
          // 获取预编译语句对象
          ps = conn.prepareStatement(sql)
          // 给SQL设置参数值
          ps.setString(1, datetime)
          ps.setInt(2, locationId)
          ps.setString(3, provinceShortName)
          ps.setInt(4, currentConfirmedCount)
          ps.setInt(5, confirmedCount)
          ps.setInt(6, suspectedCount)
          ps.setInt(7, curedCount)
          ps.setInt(8, deadCount)

          ps.executeUpdate()
        }
      })
      .outputMode("append")
      .trigger(Trigger.ProcessingTime(0))
      .option("truncate", value = false)
      .start()


    //result3.writeStream
    //  .format("console")
    //  .outputMode("complete")
    //  .trigger(Trigger.ProcessingTime(0))
    //  .option("truncate", value = false)
    //  .start()
    ////.awaitTermination()
    // +--------+-------------+--------------+--------------+----------+---------+
    // |dateId  |confirmedIncr|confirmedCount|suspectedCount|curedCount|deadCount|
    // +--------+-------------+--------------+--------------+----------+---------+
    // |20210602|24           |112458        |1821          |98916     |4995     |
    // |20210513|36           |103938        |1821          |98540     |4858     |
    // |20200519|2            |84505         |1476          |79715     |4645     |
    // |20210303|27           |101995        |1803          |96678     |4845     |

    val result3Sql =
      """
        |replace into covid19_3 (`dateId`, `confirmedIncr`, `confirmedCount`, `suspectedCount`, `curedCount`, `deadCount`)
        |values (?, ?, ?, ?, ?, ?)
        |""".stripMargin

    result3.writeStream
      .foreach(new BasicJdbcSink(result3Sql) {
        override def realProcess(sql: String, row: Row): Unit = {
          // java.lang.ClassCastException: java.lang.Integer cannot be cast to java.lang.String
          val dateId = row.getAs[String]("dateId")
          val confirmedIncr = row.getAs[Long]("confirmedIncr")
          val confirmedCount = row.getAs[Long]("confirmedCount")
          val suspectedCount = row.getAs[Long]("suspectedCount")
          val curedCount = row.getAs[Long]("curedCount")
          val deadCount = row.getAs[Long]("deadCount")
          // 获取预编译语句对象
          ps = conn.prepareStatement(sql)
          // 给SQL设置参数值
          ps.setString(1, dateId)
          ps.setLong(2, confirmedIncr)
          ps.setLong(3, confirmedCount)
          ps.setLong(4, suspectedCount)
          ps.setLong(5, curedCount)
          ps.setLong(6, deadCount)

          ps.executeUpdate()
        }
      })
      .outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))
      .option("truncate", value = false)
      .start()
    //.awaitTermination()


    //result4.writeStream
    //  .format("console")
    //  .outputMode("complete")
    //  .trigger(Trigger.ProcessingTime(0))
    //  .option("truncate", value = false)
    //  .start()
    //  //.awaitTermination()

    // +----------+-----------------+------+--------------+
    // |datetime  |provinceShortName|pid   |confirmedCount|
    // +----------+-----------------+------+--------------+
    // |2021-07-02|上海              |310000|1836          |
    // |2021-07-02|黑龙江            |230000|392           |
    // |2021-07-02|福建              |350000|389           |
    // |2021-07-02|陕西              |610000|379           |
    // |2021-07-02|内蒙古            |150000|289           |
    val result4Sql =
    """
      |replace into covid19_4  (`datetime`, `provinceShortName`, `pid`, `confirmedCount`) values (?, ?, ?, ?)
      |""".stripMargin

    result4.writeStream
      .foreach(new BasicJdbcSink(result4Sql) {
        override def realProcess(sql: String, row: Row): Unit = {
          val datetime = row.getAs[String]("datetime")
          val provinceShortName = row.getAs[String]("provinceShortName")
          val pid = row.getAs[Int]("pid")
          val confirmedCount = row.getAs[Long]("confirmedCount")

          // 获取预编译语句对象
          ps = conn.prepareStatement(sql)
          // 给SQL设置参数值
          ps.setString(1, datetime)
          ps.setString(2, provinceShortName)
          ps.setInt(3, pid)
          ps.setLong(4, confirmedCount)

          ps.executeUpdate()
        }
      })
      .outputMode("complete")
      .trigger(Trigger.ProcessingTime(0))
      .option("truncate", value = false)
      .start()


    val result5Sql =
      """
        |replace into covid19_beijing (`datetime`, `locationId`, `provinceShortName`, `cityName`, `currentConfirmedCount`,
        |`confirmedCount`, `suspectedCount`, `curedCount`, `deadCount`)
        |values (?, ?, ?, ?, ?, ?, ?, ?, ?)
        |""".stripMargin

    result5.writeStream
      .foreach(new BasicJdbcSink(result5Sql) {
        override def realProcess(sql: String, row: Row): Unit = {
          val datetime = row.getAs[String]("datetime")
          val locationId = row.getAs[Int]("locationId")
          val provinceShortName = row.getAs[String]("provinceShortName")
          val cityName = row.getAs[String]("cityName")
          val currentConfirmedCount = row.getAs[Int]("currentConfirmedCount")
          val confirmedCount = row.getAs[Int]("confirmedCount")
          val suspectedCount = row.getAs[Int]("suspectedCount")
          val curedCount = row.getAs[Int]("curedCount")
          val deadCount = row.getAs[Int]("deadCount")
          // 获取预编译语句对象
          ps = conn.prepareStatement(sql)
          // 给SQL设置参数值
          ps.setString(1, datetime)
          ps.setInt(2, locationId)
          ps.setString(3, provinceShortName)
          ps.setString(4, cityName)
          ps.setInt(5, currentConfirmedCount)
          ps.setInt(6, confirmedCount)
          ps.setInt(7, suspectedCount)
          ps.setInt(8, curedCount)
          ps.setInt(9, deadCount)

          ps.executeUpdate()
        }
      })
      .outputMode("append")
      .trigger(Trigger.ProcessingTime(0))
      .option("truncate", value = false)
      .start()
      .awaitTermination()
  }
}